package com.bins.cloud.stream.mq;

import com.alibaba.cloud.stream.binder.rocketmq.constant.RocketMQConst;
import com.bins.cloud.stream.eo.mq.MqMsg;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;

import java.util.HashMap;
import java.util.Map;

@Slf4j
@Configuration
public class CustomTransactionProducer {

    @Autowired
    private StreamBridge streamBridge;

    @Bean
    public ApplicationRunner transactionProducer() {
        return args -> {
            for (int i = 1; i <= 4; i++) {
                MqMsg mqMsg = new MqMsg();
                mqMsg.setId(String.valueOf(i));
                mqMsg.setName(String.valueOf(i));
                MessageBuilder builder = MessageBuilder.withPayload(mqMsg);
                builder.setHeader("test", String.valueOf(i)).setHeader(
                        MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
                builder.setHeader(RocketMQConst.USER_TRANSACTIONAL_ARGS, "binder");
                Message<MqMsg> msg = builder.build();
                streamBridge.send("transactionProducer-out-0", msg);
                System.out.println("send Msg:" + msg.toString());
            }
        };
    }

    @Bean
    public ApplicationRunner delayProducer() {
        return args -> {
            for (int i = 0; i < 100; i++) {
                String key = "KEY" + i;
                Map<String, Object> headers = new HashMap<>();
                headers.put(MessageConst.PROPERTY_KEYS, key);
                headers.put(MessageConst.PROPERTY_ORIGIN_MESSAGE_ID, i);
                headers.put(MessageConst.PROPERTY_DELAY_TIME_LEVEL, 2);
                headers.put("a", "userProperties");
                MqMsg mqMsg = new MqMsg();
                mqMsg.setId(String.valueOf(i));
                mqMsg.setName("Delay Message");
                Message<MqMsg> msg = new GenericMessage(mqMsg, headers);
                streamBridge.send("delayProducer-out-0", msg);
            }
        };
    }

}
